package org.obm.cassandra.client.tests;

import static me.prettyprint.cassandra.utils.StringUtils.bytes;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

import me.prettyprint.cassandra.service.Keyspace;

import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.obm.bigg.model.Folders;
import org.obm.cassandra.client.CassHelper;

public class FoldersTests extends CassandraTest {

	public void testGetSlice() throws Exception {
		Keyspace keyspace = client.getKeyspace("bigg");
		for (int i = 0; i < 100; i++) {
			ColumnPath cp = CassHelper.cp(Folders.CF, "ts-"
					+ System.nanoTime());
			keyspace.insert("INBOX", cp, bytes("conv_" + i));
		}

		// get value
		ColumnParent clp = new ColumnParent(Folders.CF);
		SliceRange sr = new SliceRange(new byte[0], new byte[0], false, 150);
		SlicePredicate sp = new SlicePredicate();
		sp.setSlice_range(sr);
		List<Column> cols = keyspace.getSlice("INBOX", clp, sp);

		assertNotNull(cols);
		for (Column c : cols) {
			System.err.println("col " + new String(c.getName()) + " -> "
					+ new String(c.getValue()));
		}
		System.err.println("cols size: " + cols.size());

		ColumnPath rm = new ColumnPath(Folders.CF);
		keyspace.remove("INBOX", rm);
	}

	public void testGetSliceWithHelper() throws Exception {
		Keyspace keyspace = client.getKeyspace("bigg");

		HashMap<String, String> toInsert = new HashMap<String, String>();
		Random r = new Random();
		for (int i = 0; i < 10; i++) {
			toInsert.put("ts-" + System.nanoTime(), "conv-"
					+ r.nextInt(Integer.MAX_VALUE));
		}

		CassHelper.insert(keyspace, Folders.CF, "INBOX", toInsert);

		// get value
		Map<String, String> fetched = CassHelper.fetchAll(keyspace, Folders.CF,
				"INBOX");
		assertNotNull(fetched);
		assertEquals(toInsert.size(), fetched.size());

		ColumnPath rm = new ColumnPath(Folders.CF);
		keyspace.remove("INBOX", rm);
	}

	public void testGetSlicePerf() throws Exception {
		Keyspace keyspace = client.getKeyspace("bigg");
		for (int i = 0; i < 100; i++) {
			ColumnPath cp = CassHelper.cp(Folders.CF, "ts-"
					+ System.nanoTime());
			keyspace.insert("INBOX", cp, bytes("conv_" + i));
		}

		// get value
		ColumnParent clp = new ColumnParent(Folders.CF);
		SliceRange sr = new SliceRange(new byte[0], new byte[0], false, 150);
		SlicePredicate sp = new SlicePredicate();
		sp.setSlice_range(sr);

		// JIT warm-up
		for (int i = 0; i < COUNT; i++) {
			keyspace.getSlice("INBOX", clp, sp);
		}

		long time = System.currentTimeMillis();
		for (int i = 0; i < COUNT; i++) {
			keyspace.getSlice("INBOX", clp, sp);
		}
		time = System.currentTimeMillis() - time;
		System.out.println("done " + COUNT + "times in " + time + "ms. "
				+ (COUNT * 1000) / time + " per sec.");

		ColumnPath rm = new ColumnPath(Folders.CF);
		keyspace.remove("INBOX", rm);
	}

}
